package com.lmax.disruptor.custom.demo;

import com.lmax.disruptor.BatchEventProcessor;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.SequenceBarrier;
import com.lmax.disruptor.WaitStrategy;
import com.lmax.disruptor.YieldingWaitStrategy;

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public class EventProcessorMain
{
    public static void main(final String[] args) throws InterruptedException, ExecutionException
    {
        int BUFFER_SIZE = 1024;
        int THREAD_NUMBERS = 4;
        /*
         * createSingleProducer创建一个单生产者的RingBuffer，
         * 第一个参数叫EventFactory，从名字上理解就是"事件工厂"，其实它的职责就是产生数据填充RingBuffer的区块。
         * 第二个参数是RingBuffer的大小，它必须是2的指数倍 目的是为了将求模运算转为&运算提高效率
         * 第三个参数是RingBuffer的生产都在没有可用区块的时候(可能是消费者（或者说是事件处理器） 太慢了)的等待策略
         */
        EventFactory<Trade> eventFactory = Trade::new;
        WaitStrategy waitStrategy = new YieldingWaitStrategy();
        final RingBuffer<Trade> ringBuffer = RingBuffer.createSingleProducer(eventFactory, BUFFER_SIZE, waitStrategy);
        //创建一个线程池
        ExecutorService executors = Executors.newFixedThreadPool(THREAD_NUMBERS);
        //创建SequenceBarrier
        SequenceBarrier sequenceBarrier = ringBuffer.newBarrier();
        //创建消息处理器
        TradeHandler tradeHandler = new TradeHandler("1");
        TradeHandler tradeHandler2 = new TradeHandler("2");
        BatchEventProcessor<Trade> transProcessor = new BatchEventProcessor<Trade>(ringBuffer, sequenceBarrier, tradeHandler);
        BatchEventProcessor<Trade> transProcessor1 = new BatchEventProcessor<Trade>(ringBuffer, sequenceBarrier, tradeHandler);
        BatchEventProcessor<Trade> transProcessor2 = new BatchEventProcessor<Trade>(ringBuffer, sequenceBarrier, tradeHandler2);
        //这一步的目的是把消费者的位置信息引用注入到生产者 如果只有一个消费者的情况可以省略
//        ringBuffer.addGatingSequences(transProcessor.getSequence());
//        ringBuffer.addGatingSequences(transProcessor.getSequence(), transProcessor1.getSequence());
        ringBuffer.addGatingSequences(transProcessor.getSequence(), transProcessor2.getSequence());
        //把消息处理器提交到线程池
        executors.submit(transProcessor, transProcessor2);
        //如果存在多个消费者，那么重复执行上面三行代码，把TradeHandler换成其他消费者类
        Future<?> future = executors.submit((Callable<Trade>) () ->
        {
            long seq;
            for (int i = 0; i < 10; i++)
            {
                seq = ringBuffer.next(); //占一个坑-----ringBuffer一个可用区块
                ringBuffer.get(seq).setPrice(Math.random() * 9999); //给这个区块放入数据
                ringBuffer.publish(seq); //发布这个区块的数据使handler(consumer)可见
                System.out.println("producer---" + seq);
            }
            return null;
        });

        future.get(); //等待生成者结束
        while (true)
        {
            // main thread not stop
        }
//        Thread.sleep(1000);//等待一秒，等消费者处理完成
//        transProcessor.halt();//通知事件（或者说消息）处理器，可以结束了（并不是马上结束）
//        executors.shutdown();//终止线程
    }
}